package yunjiao.javatutorials.apache.kafka.component;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import yunjiao.javatutorials.apache.kafka.model.InventoryUpdateEvent;
import yunjiao.javatutorials.apache.kafka.model.NotificationEvent;
import yunjiao.javatutorials.apache.kafka.model.OrderCreatedEvent;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

/**
 * 消息生产者服务
 *
 * @author yangyunjiao
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class KafkaMessageProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Value("${app.kafka.topics.order-created}")
    private String orderCreatedTopic;

    @Value("${app.kafka.topics.inventory-update}")
    private String inventoryUpdateTopic;

    @Value("${app.kafka.topics.notification}")
    private String notificationTopic;

    public void sendOrderCreatedEvent(OrderCreatedEvent event) {
        try {
            event.setMessageId(UUID.randomUUID().toString());
            event.setTimestamp(LocalDateTime.now());
            event.setSource("order-service");

            CompletableFuture<SendResult<String, Object>> future =
                    kafkaTemplate.send(orderCreatedTopic, event.getOrderId(), event);

            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info(">>> Order created event sent successfully: orderId={}, offset={}",
                            event.getOrderId(), result.getRecordMetadata().offset());
                } else {
                    log.error("Failed to send order created event: orderId={}",
                            event.getOrderId(), ex);
                    // 这里可以添加重试逻辑或告警
                }
            });

        } catch (Exception e) {
            log.error("Error sending order created event: orderId={}", event.getOrderId(), e);
            throw new RuntimeException("Failed to send Kafka message", e);
        }
    }

    public void sendInventoryUpdateEvent(InventoryUpdateEvent event) {
        try {
            event.setMessageId(UUID.randomUUID().toString());
            event.setTimestamp(LocalDateTime.now());
            event.setSource("order-service");

            CompletableFuture<SendResult<String, Object>> future =
                    kafkaTemplate.send(inventoryUpdateTopic, event.getProductId(), event);

            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info(">>> Inventory update event sent: productId={}",
                            event.getProductId());
                } else {
                    log.error("Failed to send inventory update: productId={}",
                            event.getProductId(), ex);
                    // 这里可以添加重试逻辑或告警
                }
            });

        } catch (Exception e) {
            log.error("Error sending inventory update event: productId={}",
                    event.getProductId(), e);
        }
    }

    public void sendNotificationEvent(NotificationEvent event) {
        try {
            event.setMessageId(UUID.randomUUID().toString());
            event.setTimestamp(LocalDateTime.now());
            event.setSource("order-service");

            CompletableFuture<SendResult<String, Object>> future =
                    kafkaTemplate.send(notificationTopic, event.getUserId(), event);

            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info(">>> Notification event sent: userId={}, type={}",
                            event.getUserId(), event.getType());
                } else {
                    log.error("Failed to send notification: userId={}",
                            event.getUserId(), ex);
                    // 这里可以添加重试逻辑或告警
                }
            });

        } catch (Exception e) {
            log.error("Error sending notification event: userId={}", event.getUserId(), e);
        }
    }
}
